package platform.service.ctx.ctxServer;

import com.alibaba.fastjson.JSONObject;
import common.struct.CtxServiceResult;
import common.struct.SensorData;
import common.struct.ConstantEnum.SensorDataType;
import platform.app.AppMgr;
import platform.pubsub.Publisher;
import platform.service.ctx.ctxChecker.CheckerStarter;
import platform.service.ctx.item.Item;
import platform.service.ctx.item.ItemManager;
import platform.service.ctx.pattern.PatternManager;
import platform.service.ctx.rule.RuleManager;
import platform.service.ctx.rule.resolver.ResolverType;
import platform.service.ctx.statistics.ServerStatistics;

import java.util.AbstractMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.ReentrantLock;

public class AppCtxServer extends AbstractCtxServer{

    class Processor implements Runnable {
        @Override
        public void run() {
            while(true){
                if(Thread.interrupted()){
                    return;
                }
                //如果没有待处理的msg，则跳过
                if(msgBuffer.isEmpty())
                    continue;

                //如果无法获取到锁，说明正在reset/stop
                if(!resetLock.tryLock())
                    continue;

                try{
                    Map.Entry<String, String> msgEntry = msgBuffer.poll();
                    assert msgEntry != null;
                    String channel = msgEntry.getKey();
                    String msg = msgEntry.getValue();
                    logger.debug(appMgr.getAppName() + "-CtxServer process: " + msg);

                    SensorData sensorData = SensorData.fromJSONString(msg);
                    assert sensorData.getType() == SensorDataType.MSG;
                    Item item = itemManager.addItem(channel, sensorData);
                    serverStatistics.increaseReceivedMsgNum();
                    assert appMgr.isCtxServerOn();
                    checker.check(item.getContext());
                } finally {
                    resetLock.unlock();
                }
            }
        }
    }

    private final AppMgr appMgr;
    private final ReentrantLock resetLock = new ReentrantLock();

    public AppCtxServer(AppMgr appMgr) {
        this.appMgr = appMgr;
    }

    public CtxServiceResult getResult() {
        return this.getServerStatistics().getResult();
    }

    @Override
    public void init() {
        this.msgBuffer = new ConcurrentLinkedQueue<>();
        this.patternManager = new PatternManager(this);
        this.patternManager.buildPatterns(appMgr.getPatternFile(), appMgr.getMfuncFile());
        this.ruleManager = new RuleManager(this, ResolverType.IN_TIME);
        this.ruleManager.buildRules(appMgr.getRuleFile(), null);
        this.itemManager = new ItemManager(this);
        this.changeGenerator = new ChangeGenerator(this);
        this.checker = new CheckerStarter(this, appMgr.getBfuncFile(), appMgr.getCtxValidator().toString());
        this.ctxFixer = new CtxFixer(this);
        this.serverStatistics = new ServerStatistics();

        this.processor = new Thread(new Processor(), getClass().getName() + "-Processor");
        this.processor.start();
    }

    public void reset(){
        resetLock.lock();
        //停止发送数据
        sender.interrupt();
        while(sender.isInterrupted());

        //停止处理数据
        processor.interrupt();
        while(processor.isInterrupted());

        try {
            this.patternManager.reset(appMgr.getPatternFile(), appMgr.getMfuncFile());
            this.ruleManager.reset(appMgr.getRuleFile(), null, ResolverType.IN_TIME);
            this.itemManager.reset();
            this.changeGenerator.reset();
            this.ctxFixer.reset();
            this.checker = new CheckerStarter(this, appMgr.getBfuncFile(), appMgr.getCtxValidator().toString());
            this.serverStatistics = new ServerStatistics();
            this.msgBuffer.clear();
            this.restart();

            this.processor = new Thread(new Processor(), getClass().getName() + "-Processor");
            this.processor.start();
        } finally {
            resetLock.unlock();
        }
    }

    public void stop(){
        resetLock.lock();
        //停止发送数据
        sender.interrupt();
        while(sender.isInterrupted());
        //停止处理数据
        processor.interrupt();
        while(processor.isInterrupted());

        try{
            this.patternManager = null;
            this.ruleManager = null;
            this.itemManager = null;
            this.checker = null;
            this.changeGenerator = null;
            this.ctxFixer = null;
            this.serverStatistics = null;
            this.msgBuffer = null;
        } finally {
            resetLock.unlock();
        }
    }


    @Override
    public void onMessage(String channel, String msg) {
        if(msg == null){
            return;
        }

        this.msgBuffer.offer(new AbstractMap.SimpleEntry<>(channel, msg));
    }


    @Override
    public void run() {
        while(true){
            if(Thread.interrupted()){
                return;
            }
            for(String channel : itemManager.getChannel2IndexQue().keySet()){
                ConcurrentLinkedQueue<Long> indexQue = itemManager.getChannel2IndexQue().get(channel);
                if(indexQue.isEmpty()){
                    continue;
                }
                long sendIndex = indexQue.peek();
                if(!itemManager.getValidatedItemMap().containsKey(sendIndex)){
                    continue;
                }
                Item originalItem = itemManager.getItem(sendIndex);
                Item validatedItem = itemManager.getValidatedItem(sendIndex);
                List<JSONObject> pubMessageList = itemManager.buildValidatedMessageList(validatedItem, originalItem);
                for(JSONObject pubMessage : pubMessageList){
                    logger.debug(appMgr.getAppName() + "-CtxServer pub " + pubMessage.toJSONString()  +" from " + channel + " to " + appMgr.getAppName());
                    Publisher.publish(channel, appMgr.getGrpId(), 0, pubMessage.toJSONString());
                    serverStatistics.increaseSentMsgNum();
                }
                itemManager.removeValidatedItem(sendIndex);
                //itemManager.removeItem(sendIndex);
                indexQue.poll();
            }
        }
    }
}
